package cn.edu.jxust.core.infrastructure.mqtt;

import cn.edu.jxust.core.infrastructure.mqtt.handler.MessageHandler;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;

/**
 * @author liuguangyun
 * @version 1.0
 * @description
 * @since 2020/07/01 15:10
 */
@Slf4j
@SuppressWarnings("all")
public class PushCallback implements MqttCallback {

    /**
     * 自己创建的 MqClient 对象
     */
    private final MqClient mqClient;

    /**
     * 重新连接的次数
     */
    private int reConnTimes = 0;

    /**
     * 消息处理类
     */
    private final MessageHandler handler;

    /**
     * 构造函数
     * @param client MqClient
     * @param handler MessageHandler
     */
    public PushCallback(MqClient client, MessageHandler handler) {
        this.mqClient = client;
        this.handler = handler;
    }

    /**
     * 掉线时调用函数
     * @param cause Throwable
     */
    @Override
    public void connectionLost(Throwable cause) {

        log.info("Mqtt 连接终端, 尝试重新连接, 断开原因: " + cause.getMessage());
        cause.printStackTrace();
        while (!mqClient.reConnect()) {
            try {
                if (reConnTimes++ > 20) {//前20次每秒重连一次
                    Thread.sleep(1000);
                } else {//超过20次后没10s重连一次
                    Thread.sleep(10000);
                }
            } catch (InterruptedException e) {
                log.error(e.getMessage());
            }
        }
        //将重连次数置为0
        reConnTimes = 0;

    }

    /**
     * paho 本身有个bug，即此函数接口抛出异常都会回调到connectionLost()的接口，故需要在此函数中用try catch包起来处理，
     *   确保无异常抛出。
     * @param topic topic
     * @param message message
     */
    @Override
    public void messageArrived(String topic, MqttMessage message)  {

        try {
           // 数据 protobuf 解码
           byte[] payload = message.getPayload();
           OnenetMq.Msg obj = OnenetMq.Msg.parseFrom(payload);
           handler.handle(obj.getMsgid(), new String(obj.getData().toByteArray()));
       }catch(Exception e){
           log.error("PushCallBack --> messageArrived 数据解析异常: {}", e.getMessage());
       }finally {
           if(mqClient.getManualAcks()){
               mqClient.messageArrivedComplete(message);
           }
       }

    }


    /**
     * @param token IMqttDeliveryToken
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {

        IMqttAsyncClient client = token.getClient();
        log.info("deliveryComplete: {}", client);

    }

}
